package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/18 0018 19:57
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * Flink对迟到数据的处理:
 * *  1. 推迟水位线的推进(设定乱序时间,能兼顾大多数的迟到数据)
 * *        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
 * *  2. 延迟窗口的关闭（兼顾迟到很久的数据）
 * *       .allowedLateness(Time.minutes(1))
 * *       窗口会在水位线推进到窗口的结束时间开始计算，但是窗口不会关闭， 会再等待1分钟， 接下来来的数据，还可以正常
 * *       进入该窗口，只要有新的数据到来，窗口会再次计算。
 * *  3. 输出到侧输出流（漏网之鱼）
 * *      .sideOutputLateData(lateOutputTag)
 * <p>
 * 结果：说明虽然水位线延迟了2秒，每次窗口的计算延迟了2秒，
 * 但是0-9秒的数据会进入到一个窗口，10-19秒的数据会进入到另一个窗口
 * input> Event(user=tom, url=/home, ts=1000)
 * input> Event(user=tom, url=/home, ts=5000)
 * input> Event(user=tom, url=/home, ts=10000)
 * input> Event(user=tom, url=/home, ts=11000)
 * input> Event(user=tom, url=/home, ts=12000)
 * sum> (tom,2)
 * input> Event(user=tom, url=/home, ts=15000)
 * input> Event(user=tom, url=/home, ts=22000)
 * sum> (tom,4)
 */
public class Flink10_LateData {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        ds.print("input");

        //每10秒统计每个用户的点击次数
        OutputTag<Tuple2<String, Integer>> lateOutputTag = new OutputTag<Tuple2<String, Integer>>("late") {
        };

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumDs = ds.map(
                        event -> Tuple2.of(event.getUser(), 1)
                ).returns(
                        Types.TUPLE(Types.STRING, Types.INT)
                ).keyBy(
                        t -> t.f0
                ).window(
                        TumblingEventTimeWindows.of(Time.seconds(10))
                )

                /**
                 * 延迟窗口的关闭
                 *
                 * input> Event(user=tom, url=/home, ts=0)
                 * input> Event(user=tom, url=/home, ts=1)
                 * input> Event(user=tom, url=/home, ts=12000)
                 * sum> (tom,2)
                 * input> Event(user=tom, url=/home, ts=3)
                 * sum> (tom,3)
                 * input> Event(user=tom, url=/home, ts=6)
                 * sum> (tom,4)
                 *
                 * 想要关闭，需要推进水位线到1分钟，然后再迟到的数据，
                 * 会被放到侧流
                 *
                 * input> Event(user=tom, url=/home, ts=0)
                 * input> Event(user=tom, url=/home, ts=1)
                 * input> Event(user=tom, url=/home, ts=12000)
                 * sum> (tom,2)
                 * input> Event(user=tom, url=/home, ts=4)
                 * sum> (tom,3)
                 * input> Event(user=tom, url=/home, ts=6)
                 * sum> (tom,4)
                 * input> Event(user=tom, url=/home, ts=7)
                 * sum> (tom,5)
                 * input> Event(user=tom, url=/home, ts=13000)
                 * input> Event(user=tom, url=/home, ts=8)
                 * sum> (tom,6)
                 * input> Event(user=tom, url=/home, ts=88000)
                 * sum> (tom,2)
                 * input> Event(user=tom, url=/home, ts=6)
                 * late> (tom,1)
                 * input> Event(user=tom, url=/home, ts=8)
                 * late> (tom,1)
                 */
                .allowedLateness(
                        Time.minutes(1)
                )
                .sideOutputLateData(
                        lateOutputTag
                )
                .sum(1);

        sumDs.print("sum");

        //基于侧流处理迟到的数据
        sumDs.getSideOutput(lateOutputTag).print("late");

        env.execute();


    }

}
